package tech.realcpf.flink.catalog.store;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.catalog.CatalogStore;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.factories.CatalogStoreFactory;
import org.apache.flink.table.factories.FactoryUtil;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

import static org.apache.flink.table.factories.FactoryUtil.createCatalogStoreFactoryHelper;

public class JcCatalogStoreFactory implements CatalogStoreFactory {
    private String endPoints;
    private String prefix;
    private String user;
    private String passwd;
    @Override
    public CatalogStore createCatalogStore() {
        return new JcStore(prefix,endPoints,user,passwd);
    }

    @Override
    public void open(Context context) throws CatalogException {
        FactoryUtil.FactoryHelper<CatalogStoreFactory> factoryHelper =
                createCatalogStoreFactoryHelper(this, context);
        factoryHelper.validate();


        ReadableConfig options = factoryHelper.getOptions();
        endPoints = options.get(JcCatalogStoreFactoryOptions.END_POINTS);
        prefix = options.get(JcCatalogStoreFactoryOptions.PREFIX);
        user = options.get(JcCatalogStoreFactoryOptions.USER);
        passwd = options.get(JcCatalogStoreFactoryOptions.PASSWD);
    }

    @Override
    public void close() throws CatalogException {

    }

    @Override
    public String factoryIdentifier() {
        return JcCatalogStoreFactoryOptions.IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> options = new HashSet<>();
        options.add(JcCatalogStoreFactoryOptions.END_POINTS);
        options.add(JcCatalogStoreFactoryOptions.PREFIX);
        options.add(JcCatalogStoreFactoryOptions.USER);
        options.add(JcCatalogStoreFactoryOptions.PASSWD);
        return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        return Collections.emptySet();
    }
}
